package de.uni_stuttgart.ipvs.ids.replication;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.Collection;
import java.util.Vector;

import de.uni_stuttgart.ipvs.ids.communication.MessageWithSource;
import de.uni_stuttgart.ipvs.ids.communication.NonBlockingReceiver;
import de.uni_stuttgart.ipvs.ids.communication.ReadRequestMessage;
import de.uni_stuttgart.ipvs.ids.communication.ReleaseReadLock;
import de.uni_stuttgart.ipvs.ids.communication.ReleaseWriteLock;
import de.uni_stuttgart.ipvs.ids.communication.RequestReadVote;
import de.uni_stuttgart.ipvs.ids.communication.RequestWriteVote;
import de.uni_stuttgart.ipvs.ids.communication.ValueResponseMessage;
import de.uni_stuttgart.ipvs.ids.communication.Vote;
import de.uni_stuttgart.ipvs.ids.communication.WriteRequestMessage;

/**
 * A class realizing the majority consensus protocol for replication.<br>
 * Offers methods to read and write a value which is replicated.
 * 
 * @param <T>
 *            The type for the {@link VersionedValue} this majority consensus
 *            class handles.
 * @author Team 3/4 Elkhan Dadashov, Ahmed Labib, Dhaval A Shah, Patrick Stortz
 */
public class MajorityConsensus<T> {
	/** The replicas handle by this majority consensus. */
	protected Collection<SocketAddress> replicas;
	/** The DatagramSocket used by this majority consensus. */
	protected DatagramSocket socket;
	/** The NonBlockingReceiver used by this majority consensus. */
	protected NonBlockingReceiver nbio;
	/** The overall timeout for receiving a certain amount of messages. */
	private final static int TIMEOUT = 1000;
	/** The name identifying this majority consensus. */
	private String name;

	/**
	 * Creates a class for handling the majority consensus protocol for
	 * replication.
	 * 
	 * @param replicas
	 *            The replicas in charge for a specific {@link VersionedValue}.
	 * @throws SocketException
	 *             Thrown if the socket cannot be created / bound.
	 */
	public MajorityConsensus(Collection<SocketAddress> replicas)
			throws SocketException {
		this.replicas = replicas;
		SocketAddress address = new InetSocketAddress("127.0.0.1", 4999);
		this.socket = new DatagramSocket(address);
		this.nbio = new NonBlockingReceiver(socket);
		this.name = "Consensus @ " + socket.getLocalPort();
	}

	/** @return the name identifying this majority consensus. */
	public String getName() {
		return name;
	}

	/**
	 * Calculates the needed amount of votes to reach a read or write quorum.
	 * 
	 * @return the needed amount of votes to reach a read or write quorum.
	 */
	public int getAmountOfNeededVotes() {
		if (replicas.size() % 2 == 0) {
			return (replicas.size() / 2) + 1;
		} else {
			return (replicas.size() + 1) / 2;
		}
	}

	/**
	 * vale from the replicas.
	 * 
	 * @return the {@link VersionedValue} with the currently highest version
	 *         number.
	 * @throws QuorumNotReachedException
	 *             Thrown if the read quorum could not be reached.
	 * @throws Exception
	 *             Thrown if other unexpected errors occur.
	 */
	public VersionedValue<T> get() throws QuorumNotReachedException, Exception {

		System.out.println(this.getName()
				+ " | -------------------------------------------------------");// DEBUG

		VersionedValue<T> versionedValue = null;
		try {
			// get the read locks
			System.out.println(this.getName() + " | Getting the read locks...");// DEBUG
			Collection<MessageWithSource<Vote>> yesReplies = requestReadVote();

			// find the vote with the highest version number
			System.out.println(this.getName()
					+ " | Finding vote with highest version number within "
					+ yesReplies.size() + " YES votes...");// DEBUG
			MessageWithSource<Vote> newestVote = getVoteWithHighestVersionNumber(yesReplies);

			// read the newest value
			System.out.println(this.getName() + " | Reading newest value (VN="
					+ newestVote.getMessage().getVersion() + ") from "
					+ newestVote.getSource() + "...");// DEBUG
			T value = readReplica(newestVote.getSource());

			// check if we could read the value successfully
			if (value != null) {
				// create the VersionedValue to return
				versionedValue = new VersionedValue<T>(newestVote.getMessage()
						.getVersion(), value);
			}

			// release the read locks
			System.out.println(this.getName() + " | Releasing "
					+ yesReplies.size() + " read locks...");// DEBUG
			releaseReadLock(toSocketAddressCollection(yesReplies));

		} catch (QuorumNotReachedException e) {
			// make sure we release the read locks in case we do not achieve a
			// quorum
			try {
				if (e.achieved != null) {
					System.err.println(this.getName()
							+ " | Error: Read quorum not reached - Got only "
							+ e.getAchieved().size() + " of "
							+ getAmountOfNeededVotes() + " needed votes!");// DEBUG
					System.err.println(this.getName() + " | Releasing "
							+ e.getAchieved().size() + " read locks...");// DEBUG
					releaseReadLock(e.getAchieved());
				}
			} catch (Exception e1) {
				System.err.println(this.getName() + " | "
						+ " Exception: Releasing the read locks failed!");
				// forward the exception
				throw e1;
			}
			// forward the exception
			throw e;
		}

		System.out.println(this.getName()
				+ " | -------------------------------------------------------");// DEBUG

		return versionedValue;
	}

	/**
	 * Sets the value of the replicas.
	 * 
	 * @param value
	 *            The value to set.
	 * @throws QuorumNotReachedException
	 *             Thrown if the write quorum could not be reached.
	 * @throws Exception
	 *             Thrown if other unexpected errors occur.
	 */
	public void set(T value) throws QuorumNotReachedException, Exception {

		System.out.println(this.getName()
				+ " | -------------------------------------------------------");// DEBUG

		try {
			// get the write locks
			System.out
					.println(this.getName() + " | Getting the write locks...");// DEBUG
			Collection<MessageWithSource<Vote>> yesReplies = requestWriteVote();

			// find the newest value
			System.out.println(this.getName()
					+ " | Finding vote with highest version number within "
					+ yesReplies.size() + " YES votes...");// DEBUG
			MessageWithSource<Vote> newestVote = getVoteWithHighestVersionNumber(yesReplies);

			// write the new value with the new version number
			System.out.println(this.getName() + " | Writing new value " + value
					+ " (VN=" + (newestVote.getMessage().getVersion() + 1)
					+ ") to all locked replicas...");// DEBUG
			writeReplicas(toSocketAddressCollection(yesReplies),
					new VersionedValue<T>(
							newestVote.getMessage().getVersion() + 1, value));

			// release the write locks
			System.out.println(this.getName() + " | Releasing "
					+ yesReplies.size() + " write locks...");// DEBUG
			releaseWriteLock(toSocketAddressCollection(yesReplies));

		} catch (QuorumNotReachedException e) {
			// make sure we release the write locks in case we do not achieve a
			// quorum
			try {
				if (e.achieved != null) {
					System.err.println(this.getName()
							+ " | Error: Write quorum not reached - Got only "
							+ e.getAchieved().size() + " of "
							+ getAmountOfNeededVotes() + " needed votes!");// DEBUG
					System.err.println(this.getName() + " | Releasing "
							+ e.getAchieved().size() + " write locks...");// DEBUG
					releaseWriteLock(e.getAchieved());
				}
			} catch (Exception e1) {
				System.err.println(this.getName() + " | "
						+ " Exception: Releasing the write locks failed!");
				// forward the exception
				throw e1;
			}
			// forward the exception
			throw e;
		}

		System.out.println(this.getName()
				+ " | -------------------------------------------------------");// DEBUG
	}

	/**
	 * Requests read votes from all existing replicas.
	 * 
	 * @return A collection with the replies from the replicas which sent a YES
	 *         vote.
	 * @throws QuorumNotReachedException
	 *             Thrown if the read quorum could not be reached.
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 */
	private Collection<MessageWithSource<Vote>> requestReadVote()
			throws QuorumNotReachedException, IOException,
			ClassNotFoundException {

		// create the request message
		DatagramPacket requestReadVote = getPacket(new RequestReadVote());
		// loop through all replicas
		for (SocketAddress replicaAddress : this.replicas) {
			// send the request message to one replica
			requestReadVote.setSocketAddress(replicaAddress);
			socket.send(requestReadVote);
		}
		// receive the packets
		Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT,
				replicas.size());
		// transform the packets
		Collection<MessageWithSource<Vote>> replies = NonBlockingReceiver
				.unpack(packets);

		return checkQuorum(replies);
	}

	/**
	 * Releases the read locks of certain replicas.
	 * 
	 * @param lockedReplicas
	 *            The collection containing the SocketAddress of the replicas to
	 *            unlock.
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 * @throws Exception
	 *             Thrown if the amount of replies is not equal to the amount of
	 *             locked replicas and/or the replies of the locked replicas
	 *             contain at least one NACK.
	 */
	private void releaseReadLock(Collection<SocketAddress> lockedReplicas)
			throws IOException, ClassNotFoundException, Exception {

		String replyAmountError = null;
		// check if there are any locks to release
		if (lockedReplicas != null && lockedReplicas.isEmpty() == false) {
			// create the release message
			DatagramPacket releaseReadLock = getPacket(new ReleaseReadLock());
			// loop through all locked replicas
			for (SocketAddress lockedReplicaAddress : lockedReplicas) {
				// send the release message to one locked replica
				releaseReadLock.setSocketAddress(lockedReplicaAddress);
				socket.send(releaseReadLock);
			}

			// receive the packets
			Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT,
					lockedReplicas.size());
			// transform the packets
			Collection<MessageWithSource<Vote>> replies = NonBlockingReceiver
					.unpack(packets);

			// check if we got replies from all replicas
			if (replies.size() != lockedReplicas.size()) {
				// not all locked replicas replied with an ACK or NACK
				// there still might be replicas remaining locked
				// this should not happen, since we assumed, that the nodes are
				// not failing while being locked
				replyAmountError = "Only got ACKs/NACKs for releasing the read lock from "
						+ replies.size()
						+ " out of "
						+ lockedReplicas.size()
						+ " locked replicas!";
			} else {
				System.out.println(this.getName()
						+ " | > Successfully got all " + replies.size()
						+ " out of " + lockedReplicas.size()
						+ " ACKs/NACKs for releasing the read lock.");// DEBUG
			}

			// check if some replicas did not reply (might be still locked)
			if (replyAmountError != null) {
				throw new Exception("Error: " + replyAmountError);
			}
		}
	}

	/**
	 * Reads the value of a specific replica.
	 * 
	 * @param replica
	 *            The SocketAddress of the replica to read the value from.
	 * @return The value of the specific replica
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 * @throws Exception
	 *             Thrown if the amount of replies received is not like expected
	 *             a single reply.
	 */
	@SuppressWarnings("unchecked")
	// generic types are lost during compilation, we cannot check against it
	private T readReplica(SocketAddress replica) throws IOException,
			ClassNotFoundException, Exception {

		T value = null;

		// create the read request message
		DatagramPacket readRequest = getPacket(new ReadRequestMessage());
		// send the read request message
		readRequest.setSocketAddress(replica);
		socket.send(readRequest);

		// wait for the packets
		Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT, 1);
		// transform the packets
		Collection<MessageWithSource<ValueResponseMessage<T>>> replies = NonBlockingReceiver
				.unpack(packets);
		// check if we have only one reply (like expected)
		Object[] repliesArray = replies.toArray();
		if (repliesArray.length == 1) {
			// cast the single reply to the according MessageWithSource
			MessageWithSource<ValueResponseMessage<T>> valueResponse = ((MessageWithSource<ValueResponseMessage<T>>) repliesArray[0]);
			value = valueResponse.getMessage().getValue();
			System.out.println(this.getName()
					+ " | > Successfully read the value " + value + "!");// DEBUG
		} else {
			if (repliesArray.length == 0) {
				// we got zero replies
				// this should not happen, since we assumed, that the nodes are
				// not failing while being locked
				throw new Exception("Error: "
						+ "Got zero replies for a single read request!");
			} else {
				// we got more than one reply
				throw new Exception("Error: "
						+ "Got more than one reply for a single read request!");
			}

		}

		return value;
	}

	/**
	 * Requests write votes from all existing replicas.
	 * 
	 * @return A collection with the replies from the replicas which sent a YES
	 *         vote.
	 * @throws QuorumNotReachedException
	 *             Thrown if the read quorum could not be reached.
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 */
	private Collection<MessageWithSource<Vote>> requestWriteVote()
			throws QuorumNotReachedException, IOException,
			ClassNotFoundException {

		// create the request message
		DatagramPacket requestWriteVote = getPacket(new RequestWriteVote());
		// loop through all replicas
		for (SocketAddress replicaAddress : this.replicas) {
			// send the request message to one replica
			requestWriteVote.setSocketAddress(replicaAddress);
			socket.send(requestWriteVote);
		}
		// receive the packets
		Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT,
				replicas.size());
		// transform the packets
		Collection<MessageWithSource<Vote>> replies = NonBlockingReceiver
				.unpack(packets);

		return checkQuorum(replies);
	}

	/**
	 * Releases the write locks of certain replicas.
	 * 
	 * @param lockedReplicas
	 *            The collection containing the SocketAddress of the replicas to
	 *            unlock.
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 * @throws Exception
	 *             Thrown if the amount of replies is not equal to the amount of
	 *             locked replicas and/or the replies of the locked replicas
	 *             contain at least one NACK.
	 */
	private void releaseWriteLock(Collection<SocketAddress> lockedReplicas)
			throws IOException, ClassNotFoundException, Exception {

		String replyAmountError = null;

		// check if there are any locks to release
		if (lockedReplicas != null && lockedReplicas.isEmpty() == false) {
			// create the release message
			DatagramPacket releaseWriteLock = getPacket(new ReleaseWriteLock());
			// loop through all locked replicas
			for (SocketAddress lockedReplicaAddress : lockedReplicas) {
				// send the release message to one locked replica
				releaseWriteLock.setSocketAddress(lockedReplicaAddress);
				socket.send(releaseWriteLock);
			}

			// receive the packets
			Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT,
					lockedReplicas.size());
			// transform the packets
			Collection<MessageWithSource<Vote>> replies = NonBlockingReceiver
					.unpack(packets);

			// check if we got replies from all locked replicas
			if (replies.size() != lockedReplicas.size()) {
				// not all locked replicas replied with an ACK or NACK
				// there still might be replicas remaining locked
				// this should not happen, since we assumed, that the nodes are
				// not failing while being locked
				replyAmountError = "Only got ACKs/NACKs for releasing the write lock from "
						+ replies.size()
						+ " out of "
						+ lockedReplicas.size()
						+ " locked replicas!";
			} else {
				System.out.println(this.getName()
						+ " | > Successfully got all " + replies.size()
						+ " out of " + lockedReplicas.size()
						+ " ACKs/NACKs for releasing the write lock.");// DEBUG
			}

		}

		// check if some replicas did not reply (might be still locked)
		if (replyAmountError != null) {
			throw new Exception("Error: " + replyAmountError);
		}
	}

	/**
	 * Updates certain replicas with the new value.
	 * 
	 * @param lockedReplicas
	 *            A collection of the SocketAddress of the replicas to update.
	 * @param newValue
	 *            The new value to update
	 * @throws IOException
	 *             Thrown if there is an error during sending the messages or
	 *             receiving the replies.
	 * @throws ClassNotFoundException
	 *             Thrown if one of the object in the messages or replies cannot
	 *             be serialized/deserialized.
	 * @throws Exception
	 *             Thrown if the amount of replies received is not like expected
	 *             a single reply.
	 */
	private void writeReplicas(Collection<SocketAddress> lockedReplicas,
			VersionedValue<T> newValue) throws IOException,
			ClassNotFoundException, Exception {

		String replyAmountError = null;

		// create the write request message
		DatagramPacket writeRequest = getPacket(new WriteRequestMessage<T>(
				newValue));
		// loop through all locked replicas
		for (SocketAddress lockedReplicaAddress : lockedReplicas) {
			// send the write message to one locked replica
			writeRequest.setSocketAddress(lockedReplicaAddress);
			socket.send(writeRequest);
		}
		// wait for the packets
		Vector<DatagramPacket> packets = nbio.receiveMessages(TIMEOUT,
				lockedReplicas.size());
		// transform the packets
		Collection<MessageWithSource<Vote>> replies = NonBlockingReceiver
				.unpack(packets);

		// check if we got replies from all locked replicas
		if (replies.size() != lockedReplicas.size()) {
			// not all locked replicas replied with an ACK or NACK
			// we might not be consistent
			// this should not happen, since we assumed, that the nodes are not
			// failing while being locked
			replyAmountError = "Only got ACKs/NACKs for writing the value from "
					+ replies.size()
					+ " out of "
					+ lockedReplicas.size()
					+ " locked replicas!";
		} else {
			// all locked replicas replied with an ACK or NACK
			System.out.println(this.getName() + " | > Successfully got all ("
					+ replies.size() + ") ACKs/NACKs for writing the value.");// DEBUG
		}

		// check if some replicas did not reply (might be inconsistent)
		if (replyAmountError != null) {
			throw new Exception("Error: " + replyAmountError);
		}
	}

	/**
	 * Checks whether a sufficient number of YES vote replies were received.
	 * 
	 * @param replies
	 *            the vote replies received.
	 * @return a collection of MessageWithSource containing only the replies of
	 *         the now locked replicas (YES voters).
	 * @throws QuorumNotReachedException
	 *             Thrown if the quorum has not been reached.
	 */
	private Collection<MessageWithSource<Vote>> checkQuorum(
			Collection<MessageWithSource<Vote>> replies)
			throws QuorumNotReachedException {

		// get the needed amount og votes to reach the quorum
		int yesVotesNeeded = getAmountOfNeededVotes();

		// create the collection for the yes votes
		Collection<MessageWithSource<Vote>> yesReplies = new Vector<MessageWithSource<Vote>>();
		// loop trough all votes
		for (MessageWithSource<Vote> reply : replies) {
			// check if this is a YES vote
			if (reply.getMessage().getState() == Vote.State.YES) {
				// add this yes vote to the collection
				yesReplies.add(reply);
			}
		}

		// check if we did not reach the quorum
		if (yesReplies.size() < yesVotesNeeded) {
			// we did not reach the quorum, throw the exception with the yes
			// votes collection
			throw new QuorumNotReachedException(yesVotesNeeded,
					toSocketAddressCollection(yesReplies));
		} else {
			System.out.println(this.getName() + " | > Successfully got "
					+ yesReplies.size() + " YES votes while at least "
					+ yesVotesNeeded + " are needed!");// DEBUG
			// we reached the quorum, return only the yes votes
			return yesReplies;
		}
	}

	/**
	 * Serializes an object into the data of a DatagramPacket
	 * 
	 * @param object
	 *            the object to serialize.
	 * @return A DatagramPacket containing the serialized object in its data.
	 * @throws IOException
	 *             Thrown if there are errors during writing the stream.
	 */
	private DatagramPacket getPacket(Object object) throws IOException {
		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
		ObjectOutputStream objectOutputStream = new ObjectOutputStream(
				byteArrayOutputStream);
		objectOutputStream.writeObject(object);
		objectOutputStream.flush();
		byte[] bytes = byteArrayOutputStream.toByteArray();
		return new DatagramPacket(bytes, bytes.length);
	}

	/**
	 * Transforms a collection of MessageWithSource<Vote> to a collection of
	 * SocketAddress.
	 * 
	 * @param messagesWithSource
	 *            the collection of MessageWithSource<Vote>.
	 * @return a collection of SocketAddress containing the information of the
	 *         collection of MessageWithSource<Vote>.
	 */
	private Collection<SocketAddress> toSocketAddressCollection(
			Collection<MessageWithSource<Vote>> messagesWithSource) {

		// transform the message with source collection to a source collection
		Collection<SocketAddress> addresses = new Vector<SocketAddress>();
		for (MessageWithSource<Vote> messageWithSource : messagesWithSource) {
			addresses.add(messageWithSource.getSource());
		}
		return addresses;
	}

	/**
	 * Searches the MessageWithSource with the highest version number available.
	 * 
	 * @param votes
	 *            The collection of the MessageWithSource objects to analyze.
	 * @return the MessageWithSource with the newest value available.
	 */
	private MessageWithSource<Vote> getVoteWithHighestVersionNumber(
			Collection<MessageWithSource<Vote>> votes) {

		int newestVersion = -1;
		MessageWithSource<Vote> newestVote = null;

		// loop through the votes and search the highest version
		for (MessageWithSource<Vote> vote : votes) {
			if (vote.getMessage().getVersion() > newestVersion) {
				newestVote = vote;
				newestVersion = vote.getMessage().getVersion();
			}
		}

		if (newestVote != null) {
			System.out
					.println(this.getName()
							+ " | > Successfully found vote with the highest version number (VN="
							+ newestVote.getMessage().getVersion()
							+ ") available!");// DEBUG
		}
		return newestVote;
	}
}
